package com.vector.rabbitmqDemo.confirm.async.send;

import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.TimeoutException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

/**
 * @program: rabbitmqdemo
 * @description: 确认模式-异步-消息生产者队列
 * @author: Acerola
 * @create: 2021-09-12 21:31
 **/
public class Send {
    public static final String QUEUE_NAME = "confirm_async";

    public static void main(String[] args) throws IOException, TimeoutException {

        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("39.99.141.194");
        factory.setUsername("yeb");
        factory.setVirtualHost("/yeb");
        factory.setPassword("yeb");
        factory.setPort(5672);
        // 获取连接
        Connection connection = factory.newConnection();

        // 建立通道
        Channel channel = connection.createChannel();

        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 开启事务
        channel.confirmSelect();

        // 创建set
        final SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<Long>());

        // 监听事务
        channel.addConfirmListener(new ConfirmListener() {

            // handleNack
            @Override
            public void handleNack(long arg0, boolean arg1) throws IOException {
                // TODO Auto-generated method stub

                if (arg1) {
                    System.out.println("handleNack : " + arg1);
                    confirmSet.headSet(arg0 + 1).clear();
                } else {
                    System.out.println("handleNack : " + arg1);
                    confirmSet.remove(arg0);
                }
            }

            // 没有问题
            @Override
            public void handleAck(long arg0, boolean arg1) throws IOException {
                // TODO Auto-generated method stub

                if (arg1) {
                    System.out.println("handleAck :  " + arg1);
                    confirmSet.headSet(arg0 + 1).clear();
                } else {
                    System.out.println("handleAck :  " + arg1);
                    confirmSet.remove(arg0);
                }

            }
        });

        // 发送的数据
        String msg = "msg data";

        while (true) {
            long seqNo = channel.getNextPublishSeqNo();
            channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
            confirmSet.add(seqNo);
        }

    }

}

